// Copyright (c) 2022. China Mobile(SuZhou)Software Technology Co.,Ltd. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"): you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations
// under the License.

package server

import (
	"bufio"
	"context"
	"crypto/md5"
	"crypto/tls"
	"crypto/x509"
	"encoding/binary"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"os"
	"runtime"
	"strings"
	"sync"
	"time"

	"github.com/jackc/pgproto3/v2"
	"github.com/jackc/pgx/v4"

	"gitee.com/he3db/he3proxy/config"
	"gitee.com/he3db/he3proxy/core/errors"
	"gitee.com/he3db/he3proxy/core/golog"
	_ "gitee.com/he3db/he3proxy/core/hack"
	"gitee.com/he3db/he3proxy/mysql"
	"gitee.com/he3db/he3proxy/postgresql/hba"
)

const protocolVersionNumber = 196608 // 3.0
const sslRequestNumber = 80877103
const cancelRequestCode = 80877102
const gssEncReqNumber = 80877104
const protocolSSL = false
const defaultWriterSize = 16 * 1024
const moduleName = "CONN_PGSQL"

var clientConnMap = struct {
	sync.RWMutex
	m map[uint32]*ClientConn
}{m: make(map[uint32]*ClientConn, 0)}

//-----------------------------------------------------------------
// Handshake of PG
// 1. PG SQL will send 'SSLRequest' message firstly, and ask for if request SSL
// 2. Server will send a message 'S' if need SSL, or 'N' means no SSL
// 3. Notice that we don't support SSL now
// 4. When finish SSL handshard or receive 'N', client will send 'StartupMessage' to server
// 5. Than server handle 'StartupMessage' and whether request auth or not. If need auth will send 'AuthenticationRequest' to client
// 6. PG support muti-kind auth type. Now we support md5 first, when finish auth the server will return 'AuthenticationOk' or 'ErrorResponse'
// 7. If auth successfully, server also sent some params to client by 'ParameterStatus', contains: server_version, client_encoding and so on
// 8. At last server will send 'ReadyForQuery' to client, means connection successfully and server can handle sql requests
// 9. When client receive msg 'ReadyForQuery', it means we can run SQL operation
// for more detail information, please check this article: https://zhuanlan.zhihu.com/p/493045524
func (cc *ClientConn) handshake(ctx context.Context) error {
	m, err := cc.ReceiveStartupMessage()
	if err != nil {
		return err
	}

	switch m.(type) {
	// handle for cancel request (ctrl+c)
	case *pgproto3.CancelRequest:
		clientConnMap.RLock()
		c := clientConnMap.m[m.(*pgproto3.CancelRequest).ProcessID]
		clientConnMap.RUnlock()
		if c != nil && c.backendConn != nil && c.backendConn.ConnPg.PgConn() != nil {
			cancelRequest := &pgproto3.CancelRequest{ProcessID: c.backendConn.ConnPg.PgConn().PID(),
				SecretKey: c.backendConn.ConnPg.PgConn().SecretKey()}
			err = c.backendConn.Conn.WritePgPacket(cancelRequest.Encode(nil))
			if err != nil {
				golog.Error(moduleName, "CancelRequest", "write msg err: "+err.Error(), cc.connectionId)
				return err
			}
		}
		cc.Close()
		return err
	case *pgproto3.SSLRequest:
		return cc.handleSSLRequest(ctx)
	case *pgproto3.StartupMessage:
		return cc.handleStartupMessage(ctx, m.(*pgproto3.StartupMessage))
	default:
		err = errors.ErrFormatStr("received is not a expected packet")
		return err
	}
}

// handle client request through goroutine
func (c *ClientConn) RunPg(ctx context.Context) {
	dataSend := make([]byte, 0)
	defer func() {
		r := recover()
		if err, ok := r.(error); ok {
			const size = 4096
			buf := make([]byte, size)
			buf = buf[:runtime.Stack(buf, false)]
			golog.Error(moduleName, "Run",
				err.Error(), 0,
				"stack", string(buf))
		}
		//set transaction status for backend conn
		if c.backendConn != nil {
			if c.isInTransaction() {
				c.backendConn.IsInTransaction = true
			} else {
				c.backendConn.IsInTransaction = false
			}
		}
		c.clean()
		c.Close()
		clientConnMap.Lock()
		delete(clientConnMap.m, c.connectionId)
		clientConnMap.Unlock()
	}()
	// flag for just use master node, just use for some special cases
	// treat it simple and crude, set in transaction
	if strings.ToUpper(os.Getenv(config.SingleSession)) == "ON" {
		c.status &= ^mysql.SERVER_STATUS_AUTOCOMMIT
		c.status |= mysql.SERVER_STATUS_IN_TRANS
		c.alwaysCurNode = true
	}
	for {
		header, msg, err := c.readPacketPg()
		if err != nil {
			return
		}

		// reload configuration
		if c.configVer != c.proxy.configVer {
			err := c.reloadConfig()
			if nil != err {
				golog.Error(moduleName, "RunPg",
					err.Error(), c.connectionId,
				)
				c.writeError(err)
				return
			}
			c.configVer = c.proxy.configVer
			golog.Debug(moduleName, "RunPg",
				fmt.Sprintf("config reload ok, ver: %d", c.configVer), c.connectionId,
			)
		}

		// handle receive msg
		if err, dataSend = c.dispatchPg(ctx, header, msg, dataSend); err != nil {
			c.proxy.counter.IncrErrLogTotal()
			if err == io.EOF {
				continue
			}
			golog.Error(moduleName, "RunPg",
				err.Error(), c.connectionId,
			)
			c.writePgErr(ctx, err)
			if err == mysql.ErrBadConn {
				c.Close()
			}
			c.backendConn.Close()
		}

		if c.closed {
			return
		}

		c.pkg.Sequence = 0
	}
}

// dispatch handles client request based on command which is the first byte of the data.
// It also gets a token from server which is used to limit the concurrently handling clients.
// The most frequently used command is ComQuery.
// PostgreSQL Modified
func (cc *ClientConn) dispatchPg(ctx context.Context, header []byte, data []byte, dataSend []byte) (
	error, []byte) {
	cc.proxy.counter.IncrClientQPS()
	cmd := header[0]
	var err error
	switch cmd {
	case 'Q': /* simple query */
		simpleQuery := pgproto3.Query{}
		if err := simpleQuery.Decode(data); err != nil {
			return err, dataSend
		}
		err = cc.handleQueryPg(ctx, simpleQuery.String, append(header, data...))
		return err, dataSend
		/*
				extend query protocol, msg send in sequence
			    first phase:
			    Parse
				----------  --> return：ParseComplete
				Describe
				----------  --> return：ParameterDescription  RowDescription
				Sync
				----------  --> return：ReadyForQuery

				second phase:
				Bind
				----------  --> return：BindComplete
				Describe
				----------  --> return：RowDescription
				Execute
				----------  --> return：DataRow CommandComplete
				Sync
				----------  --> return：ReadyForQuery
				X
				end with io.EOF
		*/
	case 'P': /* parse */
		parse := pgproto3.Parse{}
		if err := parse.Decode(data); err != nil {
			return err, dataSend
		}
		sql := parse.Query
		if cc.backendConn == nil || cc.backendConn.Conn == nil {
			// parse phase will reuse connect session, if exec select first than exec insert will get an error.
			// we use temporary scheme to fix it, set session in transaction, so session will choose master node.
			// TODO but the scheme makes load balance unuseful, no suitable scheme by now. Will keep thinking.
			cc.status &= ^mysql.SERVER_STATUS_AUTOCOMMIT
			cc.status |= mysql.SERVER_STATUS_IN_TRANS
			cc.backendConn, err = cc.preHandlePg(sql, ctx)
		}
		if cc == nil || cc.backendConn == nil || err != nil {
			golog.Error(moduleName, "RunPg", err.Error(), cc.connectionId)
			return nil, dataSend
		}

		// save conn info to map, use for cancel request.
		clientConnMap.Lock()
		clientConnMap.m[cc.connectionId] = cc
		clientConnMap.Unlock()

		// packaging send msg
		dataSend = append(dataSend, header...)
		dataSend = append(dataSend, data...)
		// save parse name, use for delete parse when retrieve to connection pool
		if cc.backendConn.ParName == nil {
			cc.backendConn.ParName = make([]string, 0)
		}
		cc.backendConn.ParName = append(cc.backendConn.ParName, parse.Name)
		return nil, dataSend
	case 'B': /* bind */
		dataSend = append(dataSend, header...)
		dataSend = append(dataSend, data...)
		return nil, dataSend
	case 'E': /* execute */
		dataSend = append(dataSend, header...)
		dataSend = append(dataSend, data...)
		return nil, dataSend
	case 'F': /* fastpath function call */
	case 'C': /* close */
		c := pgproto3.Close{}
		if err := c.Decode(data); err != nil {
			return err, dataSend
		}
		err = cc.handleStmtClosePg(ctx, c)
		return err, dataSend
	case 'D': /* describe */
		dataSend = append(dataSend, header...)
		dataSend = append(dataSend, data...)
		return nil, dataSend
	case 'H': /* flush */
	//	return cc.flush(ctx)
	case 'S': /* sync */
		dataSend = append(dataSend, header...)
		dataSend = append(dataSend, data...)
		err = cc.handleStmtSyncPg(ctx, dataSend)
		dataSend = make([]byte, 0)
		return err, dataSend
	case 'X': /*Client Terminate*/
		return io.EOF, dataSend
	case 'd': /* copy data */
		dataSend = append(dataSend, header...)
		dataSend = append(dataSend, data...)
		return nil, dataSend
	case 'c': /* copy done */
		dataSend = append(dataSend, header...)
		dataSend = append(dataSend, data...)
		err = cc.handleCopy(ctx, dataSend)
		dataSend = make([]byte, 0)
		return err, dataSend
	case 'f': /* copy fail */
		dataSend = append(dataSend, header...)
		dataSend = append(dataSend, data...)
		err = cc.handleCopy(ctx, dataSend)
		dataSend = make([]byte, 0)
		return err, dataSend
	default:
		return errors.ErrFormat("command %d not supported now", cmd), dataSend
	}

	return errors.ErrFormat("command %d not supported now", cmd), dataSend
}

// handle simple query protocol
func (cc *ClientConn) handleQueryPg(ctx context.Context, sql string, data []byte) error {

	var err error
	if cc.backendConn == nil || cc.backendConn.Conn == nil {
		cc.backendConn, err = cc.preHandlePg(sql, ctx)
		if err != nil {
			golog.Error(moduleName, "handleQueryPg", err.Error(), 0, "sql", sql)
			return err
		}

		if cc != nil && cc.backendConn != nil {
			clientConnMap.Lock()
			clientConnMap.m[cc.connectionId] = cc
			clientConnMap.Unlock()
		}
	} else {
		// change status, use for load balance in begin statement.(select first and than insert/update/delete)
		// we can do load balance for 'first insert', will improve machine throughput.
		if cc.beginFlag == BEGIN_PRESTART_COMMIT {
			cc.beginFlag = BEGIN_RELSTART
		}
	}
	if cc.backendConn == nil || cc.backendConn.Conn == nil {
		return errors.ErrConnIsNil
	}
	defer cc.closeConn(cc.backendConn, false)
	golog.Debug(moduleName, "handleQueryPg",
		fmt.Sprintf("exec sql [%s] by node [%s]", sql, cc.backendConn.GetAddr()), cc.connectionId, "dbname", cc.db)

	// handle for 'begin' statement,  and when exec insert/update/delete statement add 'begin'
	// 1.begin with commit ('begin' to 'begin;...;commit')
	// 2.if select statement, will do load balance
	// 3.if insert/delete/update and so on, will add 'begin' in first statement.
	//   And set flag to BEGIN_RELSTART_BEGIN means in transaction
	// 4.deal with statement as in the transaction until commit
	if cc.beginFlag == BEGIN_PRESTART {
		err = cc.backendConn.Conn.WritePgPacket((&pgproto3.Query{String: "COMMIT;"}).Encode(data))
		if err != nil {
			golog.Error(moduleName, "handleQueryPg", fmt.Sprintf("write msg err: %s", err.Error()), cc.connectionId)
			return err
		}
		cc.beginFlag = BEGIN_PRESTART_COMMIT
	} else if cc.beginFlag == BEGIN_RELSTART {
		sqlStr := "BEGIN;" + sql
		err = cc.backendConn.Conn.WritePgPacket((&pgproto3.Query{String: sqlStr}).Encode(nil))
		if err != nil {
			golog.Error(moduleName, "handleQueryPg", fmt.Sprintf("write msg err: %s", err.Error()), cc.connectionId)
			return err
		}
		cc.beginFlag = BEGIN_RELSTART_BEGIN
	} else {
		err = cc.backendConn.Conn.WritePgPacket(data)
		if err != nil {
			golog.Error(moduleName, "handleQueryPg", fmt.Sprintf("write msg err: %s", err.Error()), cc.connectionId)
			return err
		}
	}

	err = cc.receiveBackendMsg(ctx)
	if err != nil {
		golog.Error(moduleName, "handleQueryPg", fmt.Sprintf("receiveBackend msg err: %s", err.Error()), cc.connectionId)
		return err
	}

	return nil
}

// receive server connection msg, add deal with it
func (cc *ClientConn) receiveBackendMsg(ctx context.Context) error {

	dataRes := make([]byte, 0)
readloop:
	for {

		data, err := cc.backendConn.Conn.ReadPgPacket()
		if err != nil {
			return err
		}
		// deal with copy msg
		if data[0] == 'G' || data[0] == 'W' {
			// in transaction
			cc.status &= ^mysql.SERVER_STATUS_AUTOCOMMIT
			cc.status |= mysql.SERVER_STATUS_IN_TRANS
			dataRes = append(dataRes, data...)
			cc.WriteData(dataRes)
			dataRes = make([]byte, 0)
			break readloop
		}
		if data[0] == 'H' {
			cc.status &= ^mysql.SERVER_STATUS_AUTOCOMMIT
			cc.status |= mysql.SERVER_STATUS_IN_TRANS
			dataRes = append(dataRes, data...)
			cc.WriteData(dataRes)
			dataRes = make([]byte, 0)
			continue
		}
		// add new protocol 'L' for read consistency
		if data[0] == 'L' {
			lsn := string(data[5:])
			golog.Debug("pg conn", "receiveBackendMsg", "lsn: "+lsn, cc.connectionId)
			addr := cc.backendConn.ConnPg.PgConn().Conn().RemoteAddr().String()
			if addr != "" {
				cc.nodes["node1"].NodeLsn[strings.Split(addr, ":")[0]] = lsn
			}
			continue
		}

		// deal with msg for readForQuery. return msg
		if data[0] == 'Z' {
			q := pgproto3.ReadyForQuery{}
			q.Decode(data[5:])
			// deal with 'begin-commit' statement, if begin-select will return 'T' for front,
			// means in transaction, actually backend not in transaction. Do sql with load balance
			if cc.beginFlag == BEGIN_PRESTART_COMMIT {
				data = (&pgproto3.ReadyForQuery{TxStatus: 'T'}).Encode(nil)
			} else if cc.beginFlag == BEGIN_RELSTART_BEGIN {
				cc.status &= ^mysql.SERVER_STATUS_AUTOCOMMIT
				cc.status |= mysql.SERVER_STATUS_IN_TRANS
				cc.beginFlag = BEGIN_COMMIT
			} else {
				if q.TxStatus == 'T' {
					cc.status &= ^mysql.SERVER_STATUS_AUTOCOMMIT
					cc.status |= mysql.SERVER_STATUS_IN_TRANS
				} else if q.TxStatus == 'I' && !cc.alwaysCurNode {
					cc.status |= mysql.SERVER_STATUS_AUTOCOMMIT
					cc.status &= ^mysql.SERVER_STATUS_IN_TRANS
					cc.beginFlag = BEGIN_UNSTART
				}
			}

			dataRes = append(dataRes, data...)
			cc.WriteData(dataRes)
			dataRes = make([]byte, 0)
			break readloop
		}
		dataRes = append(dataRes, data...)

		// TODO At present, all data are returned.
		// In the future, we need to consider the situation of multiple data.
		// We need to set a threshold and return in batches
		//if len(dataRes) > 16*1024 {
		//	cc.WriteData(dataRes)
		//	dataRes = make([]byte, 0)
		//}
	}
	return nil
}

// handleStmtClose handle close messages in pgsql's extended query.
func (cc *ClientConn) handleStmtClosePg(ctx context.Context, close pgproto3.Close) error {
	if cc.backendConn == nil {
		return nil
	}
	data := close.Encode(nil)
	err := cc.backendConn.Conn.WritePgPacket(data)
	if err != nil {
		golog.Error("server", "handleStmtClosePg", "write msg err: "+err.Error(), cc.connectionId)
		return err
	}
	return cc.writeCloseComplete()
}

// writeCloseComplete
func (cc *ClientConn) writeCloseComplete() error {
	closeComplete := pgproto3.CloseComplete{}
	return cc.WriteData(closeComplete.Encode(nil))
}

func (cc *ClientConn) handleStmtSyncPg(ctx context.Context, data []byte) error {
	if cc.backendConn == nil || cc.backendConn.Conn == nil {
		return nil
	}
	err := cc.backendConn.Conn.WritePgPacket(data)
	if err != nil {
		golog.Error("server", "handleStmtSyncPg", "write msg err: "+err.Error(), cc.connectionId)
		return err
	}

	err = cc.receiveBackendMsg(ctx)
	if err != nil {
		return err
	}

	return nil
}

func (cc *ClientConn) handleCopy(ctx context.Context, data []byte) error {
	if cc.backendConn == nil {
		return nil
	}
	err := cc.backendConn.Conn.WritePgPacket(data)
	if err != nil {
		golog.Error(moduleName, "handleCopy", "write msg err: "+err.Error(), cc.connectionId)
		return err
	}

	err = cc.receiveBackendMsg(ctx)
	if err != nil {
		return err
	}

	return nil
}

// ReceiveStartupMessage receives the initial connection message. This method is used of the normal Receive method
// because the initial connection message is "special" and does not include the message type as the first byte. This
// will return either a StartupMessage, SSLRequest, GSSEncRequest, or CancelRequest.
func (cc *ClientConn) ReceiveStartupMessage() (pgproto3.FrontendMessage, error) {
	header := make([]byte, 4)
	if _, err := io.ReadFull(cc.pkg.Rb, header); err != nil {
		return nil, err
	}
	msgLen := int(binary.BigEndian.Uint32(header) - 4)

	msg := make([]byte, msgLen)
	if _, err := io.ReadFull(cc.pkg.Rb, msg); err != nil {
		return nil, err
	}

	code := binary.BigEndian.Uint32(msg)

	switch code {
	case protocolVersionNumber:
		startMessage := &pgproto3.StartupMessage{}
		if err := startMessage.Decode(msg); err != nil {
			return nil, err
		}
		return startMessage, nil
	case sslRequestNumber:
		sslRequest := &pgproto3.SSLRequest{}
		if err := sslRequest.Decode(msg); err != nil {
			return nil, err
		}
		return sslRequest, nil
	case cancelRequestCode:
		cancelRequest := &pgproto3.CancelRequest{}
		if err := cancelRequest.Decode(msg); err != nil {
			return nil, err
		}
		return cancelRequest, nil
	case gssEncReqNumber:
		gssEncRequest := &pgproto3.GSSEncRequest{}
		if err := gssEncRequest.Decode(msg); err != nil {
			return nil, err
		}
		return gssEncRequest, nil
	default:
		return nil, errors.ErrFormatStr("unknown startup message code: %s", fmt.Sprint(code))
	}
}

// loadSSLCertificates
func loadSSLCertificates() (tlsConfig *tls.Config, err error) {
	tlsCert, err := tls.LoadX509KeyPair("server/certs/pgserver.crt", "server/certs/pgserver.key")
	if err != nil {
		println("Load X509 failed" + err.Error())
	}
	clientAutoPolicy := tls.RequireAndVerifyClientCert

	caCert, err := ioutil.ReadFile("server/certs/pgroot.crt")
	if err != nil {
		println("read ca file filed" + err.Error())
		return nil, nil
	}
	certPool := x509.NewCertPool()
	certPool.AppendCertsFromPEM(caCert)
	tlsConfig = &tls.Config{
		ClientAuth:   clientAutoPolicy,
		ClientCAs:    certPool,
		Certificates: []tls.Certificate{tlsCert},
	}
	return tlsConfig, nil
}

func (cc *ClientConn) handleSSLRequest(ctx context.Context) error {
	if protocolSSL {
		tlsConfig, err := loadSSLCertificates()
		if err != nil {
			return err
		}

		// 写回 'S' 表示使用 SSL 连接
		if err := cc.writeSSLRequest(ctx, 'S'); err != nil {
			return err
		}

		//将现有的连接升级为SSL连接
		if err := cc.upgradeToTLS(tlsConfig); err != nil {
			return err
		}
	} else {
		// 写回 'N' 表示不使用 SSL 连接
		if err := cc.writeSSLRequest(ctx, 'N'); err != nil {
			return err
		}
	}

	// 完成 SSL 确认后需要正式接收 StartupMessage
	m, err := cc.ReceiveStartupMessage()
	if err != nil {
		return err
	}

	msg, ok := m.(*pgproto3.StartupMessage)

	// 如果接收到的包不为启动包则报错
	if !ok {
		err := errors.ErrFormatStr("received is not a StartupMessage")
		return err
	}

	// 接收完 SSLRequest 包后接收 StartupMessage
	if err := cc.handleStartupMessage(ctx, msg); err != nil {
		return err
	}

	return nil
}

// handleStartupMessage
// Receive the client StartupMessage, obtain the client information, initialize the session and perform user authentication.
// During user authentication, the customer service terminal may disconnect and wait for the user to input the password
// before reestablishing the connection. Please note that.
// Finally, send AuthenticationOK  or ErrorResponse  to indicate the success or failure of authentication
func (cc *ClientConn) handleStartupMessage(ctx context.Context, startupMessage *pgproto3.StartupMessage) error {

	// get db username and connection database name
	cc.user = startupMessage.Parameters["user"]
	cc.db = startupMessage.Parameters["database"]
	datestyle := "ISO, MDY"
	timezone := "PRC"
	// set these envs, need pg driver pgconn support
	// github.com/jackc/pgconn/config.go
	if key, ok := startupMessage.Parameters["datestyle"]; ok {
		datestyle = key
		os.Setenv("PGDATESTYLE", key)
	}
	if key, ok := startupMessage.Parameters["timezone"]; ok {
		timezone = key
		os.Setenv("PGTZ", key)
	}
	if key, ok := startupMessage.Parameters["options"]; ok {
		os.Setenv("PGOPTIONS", key)
	}
	// set flag, means connection request by he3proxy
	// the flag default is on, set env 'HE3PROXY_FLAG' to false if you want close it.
	// connection env 'HE3PROXY' need support by db engine
	// if you use original postgres, please close this flag or will encounter an error like:
	// FATAL: unrecognized configuration parameter "he3proxy" (SQLSTATE 42704)
	if strings.ToUpper(os.Getenv(config.ConnFlag)) != "FALSE" {
		os.Setenv("HE3PROXY", "true")
	}
	if cc.db == "" {
		cc.db = cc.user
	}

	// init session and do user auth.
	if err := cc.PgOpenSessionAndDoAuth(ctx); err != nil {
		return err
	}
	// According to pgpool. default server version 9.0.0
	// Or 14 Navicat will occurs error as below.
	/*
	 SELECT d.oid, d.datname AS databasename, pg_get_userbyid(d.datdba) AS databaseowner, des.description, d.datpath,
	 d.encoding, pg_encoding_to_char(d.encoding) AS encodingname FROM pg_database d
	 LEFT JOIN pg_description des ON des.objoid = d.oid;
	 ERROR:  column d.datpath does not exist
	 LINE 1: ...byid(d.datdba) AS databaseowner, des.description, d.datpath,...
	                                                             ^
	 HINT:  Perhaps you meant to reference the column "d.datname" or the column "d.datacl".
	*/
	serverVersion := os.Getenv(config.ServerVersionEnv)
	if serverVersion == "" {
		serverVersion = "9.0.0"
	}

	parameters := map[string]string{
		"client_encoding":               "UTF8",
		"DateStyle":                     datestyle,
		"integer_datetimes":             "on",
		"is_superuser":                  "on",
		"server_encoding":               "UTF8",
		"server_version":                serverVersion,
		"TimeZone":                      timezone,
		"standard_conforming_strings":   "on",
		"default_transaction_read_only": "off",
		"in_hot_standby":                "off",
		"IntervalStyle":                 "postgres",
		"session_authorization":         cc.user,
		"application_name":              startupMessage.Parameters["application_name"],
	}

	// send ParameterStatus
	if err := cc.writeParameterStatus(parameters); err != nil {
		return err
	}

	// send BackendKeyData
	if err := cc.writeBackendKeyData(ctx, cc.connectionId); err != nil {
		return err
	}

	// send ReadyForQuery "I"
	if err := cc.writeReadyForQuery(ctx, 'I'); err != nil {
		return err
	}

	golog.Info(moduleName, "handleStartupMessage",
		fmt.Sprintf("%s connection succeeded", cc.c.RemoteAddr().String()), cc.connectionId)

	cc.pkg.Sequence = 0

	return nil
}

// writeReadyForQuery
// Param 'status' is the status code of the current backend transaction.
// "I" means idle (not in the transaction), "T" means in the transaction;
// "E" means in a failed transaction (queries will be rejected before the end of the transaction block)
// After calling this method, the cache will be cleared and all messages in the cache will be sent.
func (cc *ClientConn) writeReadyForQuery(ctx context.Context, status byte) error {

	readyForQuery := &pgproto3.ReadyForQuery{TxStatus: status}
	if err := cc.WriteData(readyForQuery.Encode(nil)); err != nil {
		return err
	}

	return nil
}

func (cc *ClientConn) writeBackendKeyData(ctx context.Context, pid uint32) error {

	backendKeyData := &pgproto3.BackendKeyData{ProcessID: pid, SecretKey: 2204724030}
	if err := cc.WriteData(backendKeyData.Encode(nil)); err != nil {
		return err
	}

	return nil
}

// writeParameterStatus
// pgAdmin require param 'client_encoding'
func (cc *ClientConn) writeParameterStatus(parameters map[string]string) error {
	for k, v := range parameters {
		parameterStatus := &pgproto3.ParameterStatus{Name: k, Value: v}
		if err := cc.WriteData(parameterStatus.Encode(nil)); err != nil {
			return errors.ErrFormat("write ParameterStatus to client failed: ", err.Error())
		}
	}
	return nil
}

// PgOpenSessionAndDoAuth
// Initialize session and perform user authentication
// There is a difference between PgSQL and mysql.
// The PgSQL client will send the password only after receiving the auth request from the server.
// At the beginning, if the MySQL client has a password, it will send the password directly to the server.
func (cc *ClientConn) PgOpenSessionAndDoAuth(ctx context.Context) error {
	var err error
	// hba check
	authConf := GetAuthenticationConfiguration()
	// Look up the method from the HBA configuration.
	//var mi methodInfo
	var hbaEntry *hba.Entry
	_, hbaEntry, err = cc.lookupAuthenticationMethodUsingRules(hba.ConnHostNoSSL, authConf)
	if err != nil {
		return err
	}
	if hbaEntry.Method.String() == "reject" {
		return errors.ErrFormat("user %s is not allowed to login.", cc.user)
	} else if hbaEntry.Method.String() == "trust" {
		if err = cc.isAceessDB(hbaEntry); err != nil {
			return err
		}
		if err = cc.writeAuthenticationOK(ctx); err != nil {
			return err
		}
		return nil
	} else if hbaEntry.Method.String() == "md5" {
		if err = cc.isAceessDB(hbaEntry); err != nil {
			return err
		}
		authData := make([]byte, 0)
		err = cc.DoAuth(ctx, authData)
		if err != nil {
			return err
		}
		return nil
	} else {
		return errors.ErrFormat("Unsupport auth method %s ", hbaEntry.Method.String())
	}
}

// DoAuth PostgreSQL
// During authentication, the client will not actively send the authentication password.
// The server needs to send different types of password authentication requests before the client returns the corresponding authentication information.
func (cc *ClientConn) DoAuth(ctx context.Context, auth []byte) error {
	var err error
	// Send an authrequest, where the MD5 encryption request is used.
	// The front end must return an MD5 encrypted password for verification.
	// Salt is randomly generated 4 bytes. Here, the first four bits of salt generated in cc are directly taken.
	salt := [4]byte{cc.salt[0], cc.salt[1], cc.salt[2], cc.salt[3]}
	authRequest := pgproto3.AuthenticationMD5Password{Salt: salt}
	//AuthenticationCleartextPassword means receive pwd without encryption.
	//authRequest := pgproto3.AuthenticationCleartextPassword{}
	if err = cc.WriteData(authRequest.Encode(nil)); err != nil {
		return errors.ErrFormat("write AuthenticationMD5Password to client failed: %s", err.Error())
	}

	// After sending the authrequest, the client (SQL shell) will exit the connection.
	// Then wait for the user to re-enter the password and then establish the connection again.
	// Currently, it only runs on the SQL shell, at least in the SQL shell.
	// Therefore, it needs to be judged when reading a byte here. When the error is EOF, it can be ended normally.
	// Read the password sent by the client.

	// format： 'p' + len + 'password' + '0'
	// length = len + password + 1
	header, msg, err := cc.readPacketPg()
	if err != nil {
		return err
	}

	if header[0] != 'p' {
		return errors.ErrFormatStr("received is not a password packet" + string(auth[0]))
	}

	// remove first byte 'p' and last end EOF. The rest is authentication information
	auth = msg[:len(msg)-1]

	//get node
	rolpassword, err := getRolPwdFromDB(cc.proxy.cfg.Nodes[0].Master, cc.user)
	if err != nil || !strings.HasPrefix(rolpassword, "md5") {
		return errors.ErrFormat("%s@%s login failed, password not correct or user not exist! time: [%s]",
			cc.user, cc.c.RemoteAddr().String(), time.Now())
	}
	// The client first performs MD5 encryption on the password entered by the user, where the user name is salt.
	// Then, the 4-bit random number md5salt sent from the server is used as salt for another MD5 encryption,
	// And sends the result to the server again as authentication information.
	res := "md5" + fmt.Sprintf("%x", md5.Sum([]byte(strings.TrimPrefix(rolpassword, "md5")+
		string([]byte{cc.salt[0], cc.salt[1], cc.salt[2], cc.salt[3]}))))

	if res != string(auth) {
		golog.Error("server", "IsAllowConnect", "error", mysql.ER_ACCESS_DENIED_ERROR,
			"ip address", cc.c.RemoteAddr().String(), " access denied by He3-Proxy.")
		return errors.ErrFormat("%s@%s login failed, password incorrect! time: [%s]", cc.user, cc.c.RemoteAddr().String(), time.Now())
	}

	if err = cc.writeAuthenticationOK(ctx); err != nil {
		return err
	}

	return nil
}

// Add the deployment node address of the he3proxy to the hba-trust, so the password field is not required.
func getRolPwdFromDB(addr string, rolname string) (string, error) {
	urlExample := "postgres://postgres:vjC2T7r!6Amf54QZ@" + addr + "/postgres"
	rolpassword := ""
	conn, err := pgx.Connect(context.Background(), urlExample)
	defer conn.Close(context.Background())
	if err != nil {
		fmt.Fprintf(os.Stderr, "Unable to connect to database: %v\n", err)
		return rolpassword, err
	}
	err = conn.QueryRow(context.Background(), "select rolpassword from pg_authid where rolname=$1", rolname).Scan(&rolpassword)
	return rolpassword, err
}

// readPacket Read general messages of postgresql protocol
func (cc *ClientConn) readPacketPg() ([]byte, []byte, error) {
	header := make([]byte, 5)
	if _, err := io.ReadFull(cc.pkg.Rb, header); err != nil {
		return nil, nil, err
	}
	msgLen := binary.BigEndian.Uint32(header[1:]) - 4
	msg := make([]byte, msgLen)
	if _, err := io.ReadFull(cc.pkg.Rb, msg); err != nil {
		return header, nil, err
	}
	return header, msg, nil
}

// writeAuthenticationOK
func (cc *ClientConn) writeAuthenticationOK(ctx context.Context) error {
	authOK := &pgproto3.AuthenticationOk{}
	if err := cc.WriteData(authOK.Encode(nil)); err != nil {
		return err
	}
	return nil
}

// writePgErr
func (cc *ClientConn) writePgErr(ctx context.Context, err error) error {
	errorResponse := &pgproto3.ErrorResponse{
		Severity:            "ERROR",
		SeverityUnlocalized: "",
		//TODO The error needs to be returned according to the error code.
		Code:    "42P04",
		Message: err.Error(),
		Detail:  "",
		Hint:    "",
	}
	if err := cc.WriteData(errorResponse.Encode(nil)); err != nil {
		return err
	}
	return cc.writeReadyForQuery(ctx, 'I')
}

// writeSSLRequest
// 'S' means agree handshake for SSL.
// 'N' means do not use SSL.
func (cc *ClientConn) writeSSLRequest(ctx context.Context, pgRequestSSL byte) error {
	if err := cc.WriteData([]byte{pgRequestSSL}); err != nil {
		return err
	}
	return nil
}

func (cc *ClientConn) WriteData(data []byte) error {
	if n, err := cc.pkg.Wb.Write(data); err != nil {
		return mysql.ErrBadConn
	} else if n != len(data) {
		return mysql.ErrBadConn
	} else {
		return nil
	}
}

func (cc *ClientConn) upgradeToTLS(tlsConfig *tls.Config) error {
	// Important: read from buffered reader instead of the original net.Conn because it may contain data we need.
	tlsConn := tls.Server(cc.c, tlsConfig)
	if err := tlsConn.Handshake(); err != nil {
		return err
	}
	cc.c = tlsConn
	cc.pkg.Wb = bufio.NewWriterSize(cc.pkg.Wb, defaultWriterSize)
	cc.tlsConn = tlsConn
	return nil
}

func (c *ClientConn) lookupAuthenticationMethodUsingRules(
	connType hba.ConnType, auth *hba.Conf,
) (mi methodInfo, entry *hba.Entry, err error) {
	var ip net.IP
	if connType != hba.ConnLocal {
		// Extract the IP address of the client.
		tcpAddr, ok := c.c.RemoteAddr().(*net.TCPAddr)
		if !ok {
			err = errors.ErrFormat("client address type %T unsupported", c.c.RemoteAddr())
			return
		}
		ip = tcpAddr.IP
	}

	// Look up the method.
	for i := range auth.Entries {
		entry = &auth.Entries[i]
		var connMatch bool
		connMatch, err = entry.ConnMatches(connType, ip)
		if err != nil {
			// TODO(knz): Determine if an error should be reported
			// upon unknown address formats.
			// See: https://github.com/cockroachdb/cockroach/issues/43716
			return
		}
		if !connMatch {
			// The address does not match.
			continue
		}
		if !entry.UserMatches(hba.MakeSQLUsernameFromPreNormalizedString(c.user)) {
			// The user does not match.
			continue
		}
		return entry.MethodFn.(methodInfo), entry, nil
	}

	// No match.
	err = errors.ErrFormat("no %s entry for host %q, user %q", *HbaConfigFile, ip, c.user)
	return
}

// Determine whether there is a configuration for accessing dB
func (cc *ClientConn) isAceessDB(hbaEntry *hba.Entry) error {
	for _, s := range hbaEntry.Database {
		if s.Value == "all" || s.Value == cc.db {
			return nil
		}
	}
	return errors.ErrFormat("user %s is not allowed to access db %s.", cc.user, cc.db)
}
